package cn.tianyigps.tianyigpskafka.task.customer;

import cn.tianyigps.tianyigpspublic.param.bo.Task;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;


/**
 * @Author: WEI.YUAN
 * @Descirption:
 * @Date: 2019/4/27_15:17
 */
@Component
@Slf4j
public abstract class NewCustomer {
    private static ObjectMapper OM = new ObjectMapper();
    /**
     * @param: Task task
     * @return
     * @date 2019/3/18 9:42
     */
    public abstract boolean checkTask(Task task);

    @KafkaListener(topics ={"${tianyi.kafka.topic}"}, containerFactory = "batchContainerFactory")
    public void receive(String json){
        if (StringUtils.isEmpty(json)) {
            log.error("kafka数据为空");
            return;
        }
        Task task = null;
        try {
            task = OM.readValue(json, Task.class);
            if(CollectionUtils.isEmpty(task.getParams())){
                log.error("kafka参数数据为空");
                return;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("task反序列化失败:{}", e);
        }
        if(!this.checkTask(task)){
            return;
        }

        consume(task);
    }

    /**
     * @param: ConsumerRecord<String,String> record
     * @date 2019/3/18 10:19
     * 消费到数据
     */
    public abstract void consume(Task task);
}
